package com.li.kafka.wechat.conf;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

@Configuration
public class KafkaConf {
  @Autowired private KafkaProperties kafkaProperties;

  @Bean
  public Producer kafkaProducer() {
    Properties prop = new Properties();
    prop.setProperty(
        ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServersConfig());
    prop.setProperty(ProducerConfig.ACKS_CONFIG, kafkaProperties.getAcksConfig());
    prop.setProperty(ProducerConfig.RETRIES_CONFIG, "0");
    prop.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, "16384");
    prop.setProperty(ProducerConfig.LINGER_MS_CONFIG, "1");
    prop.setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");
    //    prop.setProperty(
    //        ProducerConfig.PARTITIONER_CLASS_CONFIG, kafkaProperties.getPartitionerClassConfig());

    prop.setProperty(
        ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringSerializer");
    prop.setProperty(
        ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringSerializer");

    Producer<String, String> producer = new KafkaProducer(prop);
    return producer;
  }
}
